package com.atguigu.edu.app.dws;

import bean.UserRegisterActiveBean;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.datastream.AllWindowedStream;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.AllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import util.DateFormatUtil;
import util.MyClickhouseUtil;
import util.MyKafkaUtil;

import java.time.Duration;

/**
 * 统计新增用户数、活跃用户数
 */
public class DwsUserRegisterLoginWindow {
    public static void main(String[] args) throws Exception {
        //1.基本环境准备
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);
        //2.检查点相关设置

        //3.从kafka的 dwd_user_register主题、和dwd_traffic_start_log主题中读取数据
        String userRegsiter = "dwd_user_register";
        String userLongin = "dwd_traffic_unique_visitor_detail";
        String groupId = "dwd_user_register_group";
        DataStreamSource<String> regsiterDS = env.addSource(MyKafkaUtil.getKafkaConsumer(userRegsiter, groupId));
        DataStreamSource<String> loginDS = env.addSource(MyKafkaUtil.getKafkaConsumer(userLongin, groupId));

        //4.将两条流转换数据结构 jsonObj->实体类（UserRegisterActiveBean）
        SingleOutputStreamOperator<UserRegisterActiveBean> registerBeanDS = regsiterDS.process(new ProcessFunction<String, UserRegisterActiveBean>() {
            @Override
            public void processElement(String jsonStr, ProcessFunction<String, UserRegisterActiveBean>.Context ctx, Collector<UserRegisterActiveBean> out) throws Exception {
                JSONObject jsonObj = JSON.parseObject(jsonStr);
                Long ts = jsonObj.getLong("ts");
                UserRegisterActiveBean userBean = new UserRegisterActiveBean(
                        "",
                        "",
                        1L,
                        0L,
                        ts * 1000
                );
                out.collect(userBean);
            }
        });
        SingleOutputStreamOperator<UserRegisterActiveBean> loginBeanDS = loginDS.process(new ProcessFunction<String, UserRegisterActiveBean>() {
            @Override
            public void processElement(String jsonStr, ProcessFunction<String, UserRegisterActiveBean>.Context ctx, Collector<UserRegisterActiveBean> out) throws Exception {
                JSONObject jsonObj = JSON.parseObject(jsonStr);
                Long ts = jsonObj.getLong("ts");
                UserRegisterActiveBean userRegisterActiveBean = new UserRegisterActiveBean(
                        "",
                        "",
                        0L,
                        1L,
                        ts * 1000
                );
                out.collect(userRegisterActiveBean);
            }
        });
        //7.将两条流union
        DataStream<UserRegisterActiveBean> union = registerBeanDS.union(loginBeanDS);
        //指定水位线和时间戳
        SingleOutputStreamOperator<UserRegisterActiveBean> withWatermarkDS = union.assignTimestampsAndWatermarks(WatermarkStrategy.<UserRegisterActiveBean>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                .withTimestampAssigner(new SerializableTimestampAssigner<UserRegisterActiveBean>() {
                    @Override
                    public long extractTimestamp(UserRegisterActiveBean element, long recordTimestamp) {
                        return element.getTs();
                    }
                }));
        //8.开窗
        AllWindowedStream<UserRegisterActiveBean, TimeWindow> windowDS = withWatermarkDS.windowAll(TumblingEventTimeWindows.of(Time.seconds(10)));
        //9.聚合
        SingleOutputStreamOperator<UserRegisterActiveBean> reduceDS = windowDS.reduce(
                new ReduceFunction<UserRegisterActiveBean>() {
                    @Override
                    public UserRegisterActiveBean reduce(UserRegisterActiveBean value1, UserRegisterActiveBean value2) throws Exception {
                        value1.setRegisterCt(value1.getRegisterCt() + value1.getRegisterCt());
                        value1.setActiveCt(value1.getActiveCt() + value2.getActiveCt());
                        return value1;
                    }
                },
                new AllWindowFunction<UserRegisterActiveBean, UserRegisterActiveBean, TimeWindow>() {
                    @Override
                    public void apply(TimeWindow window, Iterable<UserRegisterActiveBean> input, Collector<UserRegisterActiveBean> out) throws Exception {
                        for (UserRegisterActiveBean userRegisterActiveBean : input) {
                            userRegisterActiveBean.setStt(DateFormatUtil.toYmdHms(window.getStart()));
                            userRegisterActiveBean.setEdt(DateFormatUtil.toYmdHms(window.getEnd()));
                            userRegisterActiveBean.setTs(System.currentTimeMillis());
                            out.collect(userRegisterActiveBean);
                        }

                    }
                });
        reduceDS.print(">>>>");
        //10将聚合的结果写到clickhouse
        reduceDS.addSink(MyClickhouseUtil.getSinkFunction("insert into dws_user_register_active_window values (?,?,?,?,?)"));
        env.execute();
    }
}
